package com.atguigu.PracticalProject.app.dwd.db;

import com.atguigu.PracticalProject.app.BaseSqlApp;
import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//这个表在ads层需要关联 user_info test_paper course_info test_question_info 4个维度表
public class DwdStudyExamQuestionDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdStudyExamQuestionDetail().init(5000, 2, "DwdStudyExamQuestionDetail");
    }


    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv, "DwdStudyExamQuestionDetail");

        Table testExamQuestion = tEnv.sqlQuery("" +
                " select " +
                " `data`['user_id'] user_id," +
                " `data`['exam_id'] exam_id," +
                " `data`['paper_id'] paper_id," +
                " `data`['question_id'] question_id," +
                " `data`['is_correct'] is_correct," +
                " `data`['create_time'] create_time," +
                " UNIX_TIMESTAMP(`data`['create_time']) ts" +
                " from ods_db" +
                " where `database`='edu' and `table` = 'test_exam_question'" +
                " and `type` <> 'bootstrap-start'" +
                " and `type` <> 'bootstrap-complete'" +
                " ");
        tEnv.executeSql("" +
                " create table exam_question(" +
                " user_id string," +
                " exam_id string," +
                " paper_id string," +
                " question_id string," +
                " is_correct string," +
                " create_time string," +
                " ts bigint" +
                " )" + SQLUtil.getKafkaSink("dwd_course_paper_exam_question")
        );
        testExamQuestion.executeInsert("exam_question");
    }
}
